package com.xposed.aliproxy.rxjavademo.looper.util;

import android.util.Log;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.observers.DisposableObserver;
import io.reactivex.schedulers.Schedulers;

/**
 * 轮询操作符
 * Created by Su on 2019/1/7.
 */
public class LooperUtil {
    private static final String TAG = "LooperUtil";
    private static CompositeDisposable mCompositeDisposable;

    public static void initLooper() {
        mCompositeDisposable = new CompositeDisposable();
    }

    public static void clear() {
        if (mCompositeDisposable != null)
            mCompositeDisposable.clear();
    }

    /**
     * 固定时延轮询： 近似无限次数
     */
    public static void startSimplePolling(long delay) {
        startSimplePolling(Long.MAX_VALUE, delay);
    }

    /**
     * 固定时延轮询： 指定次数
     *
     * @param count 轮询次数
     * @param delay 轮询延时 单位：ms
     */
    public static void startSimplePolling(long count, long delay) {
        Observable<Long> observable = Observable
                .intervalRange(0, count, 0, delay, TimeUnit.MILLISECONDS)
                .doOnNext(new Consumer<Long>() { // 运行线程通过subscribeOn指定
                    @Override
                    public void accept(Long aLong) throws Exception {
                        //这里使用了doOnNext，因此DisposableObserver的onNext要等到该方法执行完才会回调。
                        doWork();
                    }
                });
        DisposableObserver<Long> disposableObserver = getDisposableObserver();
        observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(disposableObserver);
        mCompositeDisposable.add(disposableObserver);
    }

    /**
     * 变长时延轮询： 指定次数
     */
    public static void startAdvancePolling() {
        Observable<Long> observable = Observable.just(0L).doOnComplete(new Action() {
            @Override
            public void run() throws Exception { // 运行线程通过subscribeOn指定
                doWork();
            }
        }).repeatWhen(new Function<Observable<Object>, ObservableSource<Long>>() {
            private long mRepeatCount;

            @Override
            public ObservableSource<Long> apply(Observable<Object> objectObservable) throws Exception {
                //必须作出反应，这里是通过flatMap操作符。
                return objectObservable.flatMap(new Function<Object, ObservableSource<Long>>() {

                    @Override
                    public ObservableSource<Long> apply(Object o) throws Exception {
                        // 取消重订阅条件
                        if (++mRepeatCount > 4) {
                            //return Observable.empty(); //发送onComplete消息，但并不会触发下游的onComplete回调。
                            return Observable.error(new Throwable("Polling work finished")); //发送onError消息，可以触发下游的onError回调。
                        }

                        // 实现两次订阅间的时延
                        return Observable.timer(3000 + mRepeatCount * 1000, TimeUnit.MILLISECONDS);
                    }
                });
            }
        });
        DisposableObserver<Long> disposableObserver = getDisposableObserver();
        observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(disposableObserver);
        mCompositeDisposable.add(disposableObserver);
    }

    /**
     * @return 观察者
     */
    private static DisposableObserver<Long> getDisposableObserver() {
        return new DisposableObserver<Long>() {

            @Override
            public void onNext(Long aLong) {
            }

            @Override
            public void onError(Throwable throwable) {
                Log.d(TAG, "DisposableObserver onError, threadId=" + Thread.currentThread().getName() + ",reason=" + throwable.getMessage());
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "DisposableObserver onComplete, threadId=" + Thread.currentThread().getName());
            }
        };
    }

    private static void doWork() {
        Log.d(TAG, "doWork start,  threadId=" + Thread.currentThread().getName());
    }
}
